package Thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 生产者和消费者模式
 * <p>
 * 　Condition是在java 1.5中才出现的，它用来替代传统的Object的wait()、notify()实现线程间的协作，相比使用Object的wait()、notify()，
 * 使用Condition的await()、signal()这种方式实现线程间协作(等待、唤醒)更加安全和高效。因此通常来说比较推荐使用Condition，阻塞队列实际上是使用了Condition来模拟线程间协作。
 * <p>
 * Condition是AQS（AbstractQueuedSynchronizer）的内部类。每个Condition对象都包含一个队列(等待队列)。等待队列是一个FIFO的队列
 * <p>
 * 等待唤醒/交替执行打印
 * Condation 的  await() 和  signal() 必须要和   lock的  lock() 和unlock()组队  要不报以下的错误：java.lang.IllegalMonitorStateException
 */
public class Condation_Study {

    private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

    private Lock lock = new ReentrantLock();

    private Condition notfull = lock.newCondition();

    private Condition notEmpty = lock.newCondition();

    public static void main(String[] args) throws InterruptedException {
        Condation_Study test = new Condation_Study();
        Producer producer = test.new Producer();//内部类创建对象
        Consumer consumer = test.new Consumer();//内部类创建对象

        new Thread(producer, "produce2:").start();
        new Thread(producer, "produce1:").start();
        new Thread(consumer, "consumer1:").start();
        new Thread(consumer, "consumer2:").start();
        new Thread(consumer, "consumer3:").start();
        new Thread(consumer, "consumer4:").start();
        new Thread(consumer, "consumer5:").start();


        Thread.sleep(10000);

    }

    /**
     * 消费消息
     */
    class Consumer implements Runnable {
        volatile boolean flag = true;

        @Override
        public void run() {
            consume();
        }

        private void consume() {
            while (flag) {
                lock.lock();
                try {
                    while (queue.isEmpty()) {
                        try {
                            System.out.println("队列空，等待数据");
                            notfull.signal();
                            notEmpty.await();
                        } catch (InterruptedException e) {
                            flag = false;
                        }
                    }

                    queue.poll();//每次移走队首元素
                    System.out.println(Thread.currentThread().getName() + "消费，队列剩余空间：" + (10 - queue.size()));
                    notfull.signalAll();
                    // notEmpty.await();
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }

                } finally {
                    lock.unlock();
                }
            }
        }
    }

    /**
     * 生产消息
     */
    class Producer implements Runnable {
        volatile boolean flag = true;

        @Override
        public void run() {
            produce();
        }

        private void produce() {
            while (flag) {
                lock.lock();//只有一个线程进来，其他线程阻塞
                try {
                    while (queue.size() == 10) {//不断循环，如果队列满了，就不能继续让生产者继续生产
                        try {
                            System.out.println("队列满，等待有空余空间");
                            notfull.await();
                        } catch (InterruptedException e) {

                            flag = false;
                        }
                    }
                    //没有满的情况下，才插入一个元素
                    queue.offer(1);
                    TimeUnit.SECONDS.sleep(1);
                    System.out.println(Thread.currentThread().getName()+"生成元素，队列剩余空间："+ (10 - queue.size()));
                    notEmpty.signalAll();      //同时通知消费者消费消息
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {

                    lock.unlock();
                }
            }


        }
    }

}









